package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Flink01_Batch_WordCount {
    public static void main(String[] args) throws Exception {
        /**
         * 1.创建Spark环境
         * 2.读文件
         * 3.将读过来的数据按照空格切分切出每一个单词
         * 4.按照单词分组，将相同的单词聚合到一块
         * 5.打印
         */

//                1.创建Flink批处理环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//                2.读文件
//        DataSource<String> dataSource = env.readFile(new TextInputFormat(new Path("input/word.txt")), "input/word.txt");
        DataSource<String> dataSource = env.readTextFile("input/word.txt");

//                3.将读过来的数据按照空格切分切出每一个单词
        FlatMapOperator<String, String> wordData = dataSource.flatMap(new FlatMapFunction<String, String>() {
            /**
             *
             * @param value 读取的每一行数据
             * @param out 返回数据 采集器 将数据采集起来发送至下游
             * @throws Exception
             */
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //1.将数据按照空格切分
                String[] words = value.split(" ");

                //2.遍历拿到每一个单词
                for (String word : words) {
                    out.collect(word);
                }
            }
        });

        //4.将单词组成Tuple2元组  第二个元素为1，为了将每个单词标记上数量，方便后面计算，其次将单词设置为第一个元素，可以方便后面按照单词聚合
        MapOperator<String, Tuple2<String, Integer>> wordToOne = wordData.map(new MapFunction<String, Tuple2<String, Integer>>() {
            /**
             *
             * @param value 经过flatMap切分后的每一个单词
             * @return 将单词组成(word, 1)返回
             * @throws Exception
             */
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
//                return new Tuple2<>(value,1);
                return Tuple2.of(value, 1);
            }
        });

//                5.按照单词分组，将相同的单词聚合到一块(spark：ReducBykey功能(1.将相同的key聚合 2.将聚合后的数据做计算))
//        wordToOne.groupBy("f0");
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOne.groupBy(0);

        //6.将聚合后的value数据做sum累加操作
        AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1);

//                6.打印
        result.print();
    }
}
